package cn.com.bluemoon.bd.flink.utils;

import org.apache.flink.api.java.utils.ParameterTool;

import java.io.InputStream;
import java.util.Properties;

/**
 * @author 刘天能
 * @createdAt 2021-06-16 13:52
 * @description kafka 配置文件加载
 */
public class KafkaPropsHolderUtils {

    private Properties props;
    private ParameterTool parameterTool;

    /**
     * 根据输入流加载配置
     *
     * @param is 输入流
     */
    public KafkaPropsHolderUtils(InputStream is) {
        try {
            parameterTool = ParameterTool.fromPropertiesFile(is);
        } catch (Exception e) {
            throw new RuntimeException("初始化kafka失败,缺少属性文件：kafka.properties", e);
        }

        try {
            props = new Properties();
            props.setProperty("bootstrap.servers", parameterTool.get("bootstrap.servers"));
            props.setProperty("group.id", parameterTool.get("group.id"));
            props.setProperty("startup.mode", parameterTool.get("startup.mode"));
            props.setProperty("start.from.timestamp", parameterTool.get("start.from.timestamp"));
        } catch (NullPointerException e) {
            throw new RuntimeException("初始化kafka失败，kafka.properties文件缺少bootstrap.servers或者group.id属性", e);
        }
    }

    /**
     * 获取cluster
     *
     * @return
     */
    public String getBootstrapServers() {
        return props.get("bootstrap.servers").toString();
    }

    /**
     * 获取 group id
     *
     * @return
     */
    public String getGroupId() {
        return props.get("group.id").toString();
    }

    /**
     * 获取数据拉取模式
     *
     * @return
     */
    public String getStartupMode() {
        return props.get("startup.mode").toString();
    }

    /**
     * 获取从时间戳启动的时间戳
     *
     * @return
     */
    public String getTimestamp() {
        return props.get("start.from.timestamp").toString();
    }

}
